package com.源码分析.flatMap等;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.Flux;

/**把流中的每一个元素转换成一个流，再把转换之后得到的所有流中的元素进行合并
 * 在逻辑上由map和flatten(就reactor-core而言，flatten类似于merge操作符)这两个操作组成。flatMap操作符的map部分将每个传入的元素转换为响应式流(T -> Flux<R>)，而flatten部分将所有
 * 生成的响应式序列合并为一个新的响应式序列，通过该序列可以传递R类型的元素
 * @author zhangsulei
 *
 */
public class FlatMapDemo {

	private Logger log = LoggerFactory.getLogger(FlatMapDemo.class);

	/**
	 *  16:09:05.705 [main] INFO  reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
		16:09:05.708 [main] INFO  reactor.Flux.Array.1 - | request(256)
		16:09:05.709 [main] INFO  reactor.Flux.Array.1 - | onNext(flux)
		16:09:05.709 [main] INFO  com.源码分析.FlatMapDemo - get:f
		16:09:05.709 [main] INFO  com.源码分析.FlatMapDemo - get:l
		16:09:05.709 [main] INFO  com.源码分析.FlatMapDemo - get:u
		16:09:05.709 [main] INFO  com.源码分析.FlatMapDemo - get:x
		16:09:05.709 [main] INFO  reactor.Flux.Array.1 - | request(1)
		16:09:05.709 [main] INFO  reactor.Flux.Array.1 - | onNext(mono)
		16:09:05.710 [main] INFO  com.源码分析.FlatMapDemo - get:m
		16:09:05.710 [main] INFO  com.源码分析.FlatMapDemo - get:o
		16:09:05.710 [main] INFO  com.源码分析.FlatMapDemo - get:n
		16:09:05.710 [main] INFO  com.源码分析.FlatMapDemo - get:o
		16:09:05.710 [main] INFO  reactor.Flux.Array.1 - | request(1)
		16:09:05.710 [main] INFO  reactor.Flux.Array.1 - | onComplete()

	 * @throws InterruptedException
	 */
	@Test
	public void test1() throws InterruptedException {
		Flux.just("flux", "mono")
			.flatMap(s -> Flux.fromArray(s.split("\\s*")))
			.subscribe(e -> log.info("get:{}", e));
		TimeUnit.SECONDS.sleep(10);
	}

	@Test
	public void test2() {
		Flux.just("user-1", "user-2", "user-3")
			.flatMap(u -> requestBooks(u).map(b -> u + "/" + b))
			.subscribe(System.out::println);
	}

	private Flux<String> requestBooks(String user) {
		return Flux.range(1, 3)
				.map(i -> "book-" + i) // 将每个数字映射为书名
				.delayElements(Duration.ofMillis(3)); // 将每本书的发送延迟一段时间，这将模拟与数据库的通信延迟
	}

}
